package Schedule::Chronic::Distributed;

=head1 NAME

Schedule::Chronic::Distributed - Constrained Distributed Scheduler for Perl

=head1 SYNOPSIS

  use Schedule::Chronic::Distributed;

  # set up scheduler
  my $chronic = new Schedule::Chronic::Distributed ( debug => 0, logtype => 'syslog', config => {%config}, pidfile_handler => $pidfile );

  # run schedule
  $chronic->schedule();

  # stop scheduled jobs on this node
  $chronic->stop_workers();

  # notify central db that node no longer running
  $chronic->update_node_availability(0);

=head1 DESCRIPTION

=head2 The Schedule::Chronic::Distributed Application

Schedule::Chronic::Distributed is a constraints-based, distributed
scheduler with a shared database backend.

This allows you to schedule times based on time, resource and other
constraints accross multiple nodes.

You can specify that a task runs exclusively when the following
conditions are met :
* a pre-requisite task changes to a specified state such as
error/complete (you can have multiple pre-requisite tasks per task)
* an interval of time has passed
* the time is between two given times in a day (and the task can run
once or in intervals in that day)


=head3 Scheduled scripts

Scheduled scripts should abide by some simple rules in order to work
smoothly with the scheduler.

There are two types of command : system and perl.

Perl commands are perl scripts that are evaluated in the worker process avoiding
the overhead of a second perl interpreter and loading/compiling dependancies on each run.
These perl scripts should therefore return 1 on success, and 0 on failure rather than exit.

System commands are shell or other scripts that are run by a forked subprocess via the
system() function, these should exit normally.

Both system and perl scripts should log any job warnings to '/tmp/schedule_task_$pid.warning',
this will be collected together with any captured stderr and stdout messages and used
to populate the task and task history. By job warnings I mean warnings specific to the purpose
of the script,i.e. "no new foo file for 24 hours", rather than "couldn't open file" which should
be handled via noticing exits and capturing stderr. $pid refers to the worker process id, in the
case of perl type scripts this will be the pid and in the case of system type scripts this will ppid.

The following code snippet will ensure that the correct pid is used for the file as system scripts will
need to use the parent pid, while perl scripts will use their own.

my $pid = ($0 eq 'chronicd') ? $$ : getppid() ;
open (WARNFILE,">/tmp/schedule_task_${pid}.warning") or die "couldn't open warning file : $!\n";
print WARNFILE "die script about to die \n";
close WARNFILE;

See examples in the ex/ directory of this distribution for example scripts

=head2 The Schedule::Chronic::Distributed Class

Class representing main scheduler. Used by daemon to schedule jobs, interacts
with all the objects in the system.


=head1 METHODS

=cut


use strict;
use Data::Dumper;

use Log::Log4perl qw(:easy);
use DateTime;
use Sys::Hostname;
use IO::Select;
use IO::Pipe;
use POSIX qw(WNOHANG);
use Tree::Binary::Dictionary;
use Proc::ProcessTable;

use Schedule::Chronic::Distributed::Task;
use Schedule::Chronic::Distributed::Requirement;
use Schedule::Chronic::Distributed::Command;
use Schedule::Chronic::Distributed::Node;

our $VERSION = 0.10;

our $running_tasks = Tree::Binary::Dictionary->new( );
our $child_sockets = Tree::Binary::Dictionary->new( );

Log::Log4perl->easy_init($INFO);

use constant CONFIG     => 0;
use constant NODE       => 1 ;
use constant MAX_RUNNING_TASKS => 2;
use constant TASK_FETCH_COUNT  => 3;
use constant FETCH_FREQUENCY   => 4;
use constant PIDFILE_HANDLER   => 5;
use constant NEXT_TASKS => 6;
use constant MASTER_PID => 7 ;
use constant TASK_PATH => 8 ;

=head2 new

  Constructor class. Creates and returns a new scheduler object.

  # set up scheduler
  my $chronic = new Schedule::Chronic::Distributed ( debug => 0, logtype => 'syslog', config => {%config}, pidfile_handler => $pidfile );

=cut


sub new {
    my ($class, %args) = @_;
    my ($node) = Schedule::Chronic::Distributed::Node->search( name => $args{config}{hostname} || hostname() );

    my @self = ();
    $self[NODE]                = $node->id;
    $self[FETCH_FREQUENCY]   ||= 10;      # 10 seconds
    $self[MAX_RUNNING_TASKS] ||= 5;
    $self[TASK_FETCH_COUNT]  ||= 10;
    $self[NEXT_TASKS]          = [];
    $self[MASTER_PID]          = $$;
    $self[PIDFILE_HANDLER]     = $args{pidfile_handler};
    $self[TASK_PATH]           = $args{config}{path};

    # FIXME: use real log conf, or set level via configuration
    return bless \@self, $class;
}


=head2 schedule

  Starts the main scheduling loop, returns when loop is exited at shutdown.

  Doesn't return anything or accept any arguments.

  # run schedule
  $chronic->schedule();

=cut

sub schedule {
  my $self = shift;
  my $logger = get_logger();
  $logger->info("started scheduling...");
  my $running_tasks_count = 0;
  my $schedule_loop_count = 0;

  while (1) {
    # check in and start timer for timeout
    $self->node->check_in($logger);
    eval {
      local $SIG{ALRM} = sub { die "schedule loop timed out after 180 seconds\n" };
      alarm 180;
      my $now = DateTime->now;
      my $wait = $now + DateTime::Duration->new( seconds => $self->[FETCH_FREQUENCY]);

      # prepare
      $logger->info("check_running_tasks if running_tasks_count : $running_tasks_count");
      check_running_tasks() if ($running_tasks_count);
      $self->get_next_tasks;

      # work through tasks
      $logger->debug("processing new tasks");
      while (my $task = $self->get_next_task) {
	warn "processing next task...\n";
	warn "task : ", $task, "status : ", $task->current_status, "\n";
	$logger->debug("updating running_task_count");
	$running_tasks_count = update_running_task_count();
	warn "running_tasks_count : $running_tasks_count / max_running_tasks : ", $self->[MAX_RUNNING_TASKS]," \n";
	if ($running_tasks_count < $self->[MAX_RUNNING_TASKS]) {
	  if ($task->current_status eq 'running') {
	    $logger->warn("task $task is already running - won't attempt to run it");
	    next;
	  }
	  $logger->debug("about to run task : $task\n");
	  my $child = $self->run_task($task);
	  $logger->debug("got pid for child : $child running task : ",$task->id);
	}
      }

      #    warn "updating running_task_count\n";
      $running_tasks_count = update_running_task_count();
      $logger->debug("now : $now / wait : $wait\n");
      while ($now < $wait) {
	sleep 2;
	$now = DateTime->now;
	if ($running_tasks_count) {
	  $logger->debug("updating running task count");
	  $running_tasks_count = update_running_task_count();
	  $logger->debug( "checking running_tasks\n");
	  check_running_tasks();
	  $logger->debug("updating running task count");
	  $running_tasks_count = update_running_task_count();
	}
      }
      alarm 0;
    };				# end of eval
    $logger->logdie("\n\n$@\n") if $@;
  }				# end of while - scheduler loop
  warn "schedule() stopping\n";
}

=head2 update_node_availability

  Object method that updates the central database of this nodes status.

  Passes the availability of the node the scheduler is currently running on.

  Doesn't return anything.

  # notify central db that node no longer running
  $chronic->update_node_availability(0);


=cut

sub update_node_availability {
    my ($self,$availability) = @_;
    my $node = $self->node;
    $node->available($availability);
    $node->update();
    return;
}

=head2 node_id

    Object method that returns the scheduler's node's id

=cut

sub node_id {
  return shift()->[NODE];
}

=head2 node

    Object method that returns the scheduler's node object.

=cut

sub node {
  my $self = shift;
  return Schedule::Chronic::Distributed::Node->retrieve($self->[NODE]);
}

=head2 pidfile_handler

   Object method that gets/sets the pidfile handler.

   This probably ought to be an internal/private method.

=cut

sub pidfile_handler {
  my ($self,$value) = @_;
  if ($value) {
    $self->[PIDFILE_HANDLER] = $value;
  }
  return $self->[PIDFILE_HANDLER];
}

=head2 am_a_child

   Object method that returns wether the current process is a child of the forking daemon parent thing.

=cut

sub am_a_child {
    my $self = shift;
    my $master_pid = $self->[MASTER_PID];
 #   warn "my pid : $$ / master pid : $master_pid\n";
    return ($$ == $master_pid ) ? 0 : 1;
}

=head2 stop_workers

  Object method to stop scheduled jobs running on this node.

  $chronic->stop_workers();


=cut

sub stop_workers {
  my $self = shift;
  my $logger = get_logger();
  warn "getting running tasks keys";
  foreach my $child_pid ($running_tasks->keys) {
    $logger->warn("stopping worker : $child_pid");
    _end_child($child_pid);
    $logger->info("get child_id task from running_tasks");
    my $task_id = $running_tasks->get($child_pid)->{task_id};
    my $task = Schedule::Chronic::Distributed::Task->retrieve($task_id);
    # child no longer around..
    $task->current_status('error');
    $task->error('chronicd scheduler shutting down at ' . DateTime->now);
    $task->_update_waiting_tasks();
    $task->notify();
    # update historical values
    $task->_update_history();
    # update current/next times
    $task->_update_times();
    $logger->info("deleting child_sockets entry and running_tasks");
    $child_sockets->delete($running_tasks->get($child_pid)->{from});
    $running_tasks->delete($child_pid);
    _end_child($child_pid);
  }
  return;
}

=head2 get_next_task

    Object method that returns the next task due to be run on this node.

=cut

sub get_next_task {
  my $self = shift;
  my $task = shift(@{$self->[NEXT_TASKS]});
  return $task;
}

=head2 get_next_tasks

    Object method that fetches the next tasks due to run.

=cut

sub get_next_tasks {
  my $self = shift;
  my $logger = get_logger();
  my $node = $self->node;
  $logger->debug("current list of tasks : ", join(', ',@{$self->[NEXT_TASKS]}));
  $logger->debug(warn "next tasks : (", join (', ',@{$self->[NEXT_TASKS]}), ")");
  unless (scalar @{$self->[NEXT_TASKS]}) {
    $logger->debug("fetching more tasks");
    my @next_tasks = Schedule::Chronic::Distributed::Task->get_next_tasks($node,$self->[TASK_FETCH_COUNT]);
    $self->[NEXT_TASKS] = \@next_tasks;
    warn "got more tasks : ", map ($_->id . ' ' . $_->name, @next_tasks), "\n";
  }
  return $self->[NEXT_TASKS];
}

=head2 check_running_tasks

   Object method that manages the tasks currently running on this node.

=cut

sub check_running_tasks {
  my $logger = get_logger();
  warn "check_running_tasks\n";
  $logger->info("checking for completed tasks");

  my $io_sel = IO::Select->new();

  $logger->info("checking children are alive and not timed out\n");
  foreach my $child_pid ($running_tasks->keys) {
    warn "checking child : $child_pid\n";
    my $running_task = $running_tasks->get($child_pid);
    # check child is still alive
    unless (kill(0, $child_pid)) {
      # child no longer around..
      $logger->info("child $child_pid is no longer running..\n");
      handle_child_ending($child_pid);
      next;
    }
    $logger->debug("child $child_pid is alive\n");

    # check child task hasn't timed out
    if ($running_task->{timeout}) {

      my $now = DateTime->now;
      my $timeout = $running_task->{started} + + DateTime::Duration->new( seconds => $running_task->{timeout});
      if ($now > $timeout) {
	# child has been running than timeout for task
	$logger->warn("child $child_pid - task ", $running_task->{task_id}, ' has timed out -- started at ', $running_task->{started});
	_end_child($child_pid, timedout => $now);
	next;
      }
      $logger->debug("child $child_pid hasn't timed out\n");
    }

    # add child listlening socket to select
    $io_sel->add($running_task->{from});
  }

  $logger->debug("checking for and handling messages from children");
  foreach my $ready ( $io_sel->can_read(1) ) {
    my $child = $child_sockets->get($ready);
    $logger->debug("child $child ready : $ready");
    if (read($ready, my $length, 4) == 4) {
      $length = unpack "L", $length;
      if (read($ready, my $message, $length) == $length) {
	$logger->debug("[parent] message : $message");
	$running_tasks->get($child)->{from}->close;
	my $ack = "ACK $child";
	$logger->debug("[parent] acknowledging child ($child) message");
	print { $running_tasks->get($child)->{to} } pack("L", length($ack)), $ack;
	$running_tasks->get($child)->{to}->close;
      }
    } else {
      ## something broken with this child...
      # FIXME: check if this ever happens, handle it properly
      $logger->warn("problem communicating with child : $child");
    }
  }
  return;
}

=head2 update_running_task_count

    function to update and return the number of tasks currently running.

=cut

sub update_running_task_count {
  my $logger = get_logger();
  ## reap children
  $logger->debug("update_running_task_count called by ", join (', ',caller));
  while ((my $child_pid = waitpid(-1, WNOHANG)) > 0) {
    $logger->debug("child_pid : $child_pid");
    handle_child_ending($child_pid) if ($running_tasks->get($child_pid));
  }
  $logger->info("running task count : ", $running_tasks->count);
  return $running_tasks->count;
}

=head2 handle_child_ending

   internal function called when child process is ending.

=cut

sub handle_child_ending {
  my $child_pid = shift;
  my $logger = get_logger();
  my $timedout = shift;
  $logger->info("handling child ending -- pid : $child_pid : / timedout : $timedout  / caller :".join(', ',caller));
  # refetch task and check/update it's status
  my $task_id = $running_tasks->get($child_pid)->{task_id};
  my $task = Schedule::Chronic::Distributed::Task->retrieve($task_id);
  unless ($task) {
    $logger->error("task : $task_id no matching object - possibly removed from db while running");
    return;
  }
  $task->remove_from_object_index();
  $task = Schedule::Chronic::Distributed::Task->retrieve($task_id); # now latest from db
  $logger->info("last run : ", $task->last_run);
  $logger->info("started : ", $running_tasks->get($child_pid)->{started});
  my $stdout_filename = "/tmp/schedule_task_${child_pid}.stdout";
  my $stderr_filename = "/tmp/schedule_task_${child_pid}.stderr";
  my $warning_filename = "/tmp/schedule_task_${child_pid}.warning";

  if ($task->current_status eq 'running' or $timedout) {
    $logger->warn("child $child_pid has died before finishing!!");
    # child died before finishing task
    open(CAPTURED_STDERR,"<$stderr_filename") or $logger->error("couldn't open $stderr_filename to capture STDERR for pid $child_pid : $!");
    my $stderr = join('',<CAPTURED_STDERR>);
    close(CAPTURED_STDERR);

    open(CAPTURED_STDOUT,"<$stdout_filename") or $logger->error("couldn't open $stdout_filename to capture STDOUT for pid $child_pid : $!");
    my $stdout = join('',<CAPTURED_STDOUT>);
    close(CAPTURED_STDOUT);

    my $error_intro = ($timedout) ? "timed out at $timedout" : 'died before completing';

    my $error_report = "Process $child_pid $error_intro \n\nSTDERR:\n$stderr\n\nSTDOUT:\n$stdout";
    $task->current_status('error');
    $task->error($error_report);
    $task->_update_waiting_tasks();
    $task->notify();
    # update historical values
    $task->_update_history();
    # update current/next times
    $task->_update_times();

  }

  unlink $stderr_filename or $logger->error("Could not unlink captured STDERR file : $stderr_filename");
  unlink $stdout_filename or $logger->error("Could not unlink captured STDOUT file : $stdout_filename");
  unlink $warning_filename or $logger->error( "couldn't remove warning file $warning_filename : $! ");
  $logger->info("[parent] $child_pid reaped");
  $logger->info("deleting child_sockets and running_tasks");
  $child_sockets->delete($running_tasks->get($child_pid)->{from});
  $running_tasks->delete($child_pid);
}

=head2 run_task

   Object method to run a task.

   Accepts task object, and forks to run task, capturing errors, etc.

=cut

sub run_task {
  my ($self,$task) = @_;
  my $logger = get_logger();
  $logger->warn("run_task called with ", join (', ',@_));
  my $node = $self->node;
  my $task_id = $task->id;
  my $to_child = IO::Pipe->new;
  my $from_child = IO::Pipe->new;
  my $now = DateTime->now;
  defined (my $child_pid = fork) or return 0; # if can't fork, try to make do

  unless ($child_pid) {
    # Child
    $to_child->reader;
    $from_child->writer;
    $from_child->autoflush(1);
    $logger->info("[child] $$ - running task : $task_id");
    die "aargh no task_id" unless ($task_id);
    my $task = Schedule::Chronic::Distributed::Task->retrieve($task_id);
    my $ran_ok = $task->run($node,$self->[TASK_PATH]);

    $logger->debug("[child] $$ - completed task $task_id ok ? : $ran_ok");
    my $message = "$$ -- $task_id completed ok : $ran_ok";
    print $from_child pack("L", length($message)), $message;
    $from_child->close;
    $logger->debug("[child] $$ - sent message");
    my $waiting_message_ack = 1;
    my $timeout = 0;
    while ($waiting_message_ack) {
      $logger->debug("[child] $$ - timeout : $timeout / waiting_message_ack : $waiting_message_ack \n");
      if ($timeout > 5) {
	$logger->error("[child] $$ - timed out waiting for response");
	last;
      }
      sleep 1;
      $logger->debug( "[child] $$ - checking for message");
      if ( read($to_child, my $length, 4) == 4 ) {
	$length = unpack "L", $length;
	if ( read($to_child, my $message, $length) == $length ) {
	  $logger->debug("[child] $$ - got message_ack : $message");
	  $waiting_message_ack = 0 if ($message eq "ACK $$");
	} else {
	  $logger->error("[child] $$ - got bad response");
	  last;
	}
      }
      $timeout++;
    }
    $logger->info("[child] $$ - exiting");
    exit 0;
  }

  # Parent
  $to_child->writer;
  $to_child->autoflush(1);
  $from_child->reader;
  $running_tasks->set($child_pid => {
				     task_id => $task_id, to => $to_child, from => $from_child,
				     started => $now, timeout => $task->timeout,
				    });
  $child_sockets->set($from_child => $child_pid );
  return $child_pid;
}


########################################
# private methods and functions

my $ppids_table;
my $pt;

sub _end_child {
  my ($pid, %args) = @_;
  my @pids = _get_pids(0,$pid);
  kill 15, @pids;
  handle_child_ending($pid,$args{timedout}) if ($args{timedout});
  my $sleep = 5;
  while ($sleep > 0) {
    last unless (kill 0, @pids);
    sleep 1;
    $sleep--;
  }
  kill 9, @pids;
}

sub _get_pids {
  my($depth,$parent) = @_;
  _build_ppids_table() unless ($depth);
  my @pids;
  foreach my $pid (@{$ppids_table->{$parent}}) {
     push (@pids, $pid, _get_pids($depth+1,$pid));
   }
  unless ($depth) {
    $ppids_table = undef;
    $pt = undef;
  }
  return @pids;
}

sub _build_ppids_table {
  $pt = Proc::ProcessTable->new;
  foreach my $proc (@{$pt->table}) {
    push (@{$ppids_table->{$proc->ppid}}, $proc->pid);
  }
}

1;

__END__


=head1 SEE ALSO

Schedule::Chronic::Distributed::Task

Schedule::Chronic::Distributed::Command

=head1 AUTHOR

aaron trevena, E<lt>teejay@droogs.orgE<gt>

=head1 COPYRIGHT AND LICENSE

Copyright (C) 2006 by Aaron Trevena

This library is free software; you can redistribute it and/or modify
it under the same terms as Perl itself, either Perl version 5.8.5 or,
at your option, any later version of Perl 5 you may have available.

=cut
